package com.ly.service.search;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.collect.Lists;
import com.google.common.primitives.Longs;
import com.ly.base.HouseSort;
import com.ly.base.RentValueBlock;
import com.ly.entity.House;
import com.ly.entity.HouseDetail;
import com.ly.entity.HouseTag;
import com.ly.entity.SupportAddress;
import com.ly.repository.HouseDetailRepository;
import com.ly.repository.HouseRepository;
import com.ly.repository.HouseTagRepository;
import com.ly.repository.SupportAddressRepository;
import com.ly.service.ServiceMultiResult;
import com.ly.service.ServiceResult;
import com.ly.service.house.IAddressService;
import com.ly.web.form.MapSearch;
import com.ly.web.form.MapSearchDistrict;
import com.ly.web.form.RentSearch;

import com.ly.web.form.SearchDistrict;
import com.sun.org.apache.xpath.internal.operations.Bool;
import org.apache.lucene.queryparser.xml.builders.BooleanQueryBuilder;
import org.elasticsearch.action.admin.indices.analyze.AnalyzeAction;
import org.elasticsearch.action.admin.indices.analyze.AnalyzeRequestBuilder;
import org.elasticsearch.action.admin.indices.analyze.AnalyzeResponse;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequestBuilder;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.geo.GeoPoint;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.query.RangeQueryBuilder;
import org.elasticsearch.index.reindex.BulkByScrollResponse;
import org.elasticsearch.index.reindex.DeleteByQueryAction;
import org.elasticsearch.index.reindex.DeleteByQueryRequestBuilder;
import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.aggregations.Aggregation;
import org.elasticsearch.search.aggregations.AggregationBuilder;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.bucket.terms.Terms;
import org.elasticsearch.search.aggregations.bucket.terms.TermsAggregationBuilder;
import org.elasticsearch.search.fetch.subphase.highlight.HighlightBuilder;
import org.elasticsearch.search.sort.SortOrder;
import org.elasticsearch.search.suggest.Suggest;
import org.elasticsearch.search.suggest.SuggestBuilder;
import org.elasticsearch.search.suggest.SuggestBuilders;
import org.elasticsearch.search.suggest.completion.CompletionSuggestionBuilder;
import org.elasticsearch.search.suggest.completion.CompletionSuggestion;
import org.modelmapper.ModelMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

/**
 * @author ：LY
 * @date ：Created in 2020/11/17 15:14
 * @modified By：
 */
@Service
public class SearchServiceImpl implements ISearchService{

    private static final Logger logger = LoggerFactory.getLogger(ISearchService.class);


    private static final String INDEX_NAME = "xunwu";

    private static final String INDEX_TYPE = "house";

    private static final String INDEX_TOPIC = "house_build";
    @Autowired
    private HouseRepository  houseRepository;

    @Autowired
    private HouseDetailRepository houseDetailRepository;

    @Autowired
    private HouseTagRepository tagRepository;

    @Autowired
    private ModelMapper modelMapper;

    @Autowired
    private TransportClient esClient;

    @Autowired
    private ObjectMapper objectMapper;
    @Autowired
    private SupportAddressRepository supportAddressRepository;

    @Autowired
    private IAddressService addressService;
    @Autowired
    private KafkaTemplate<String,String> kafkaTemplate;

    @KafkaListener(topics = INDEX_TOPIC)
    private void handleMessage(String content){
        //kafka消费者，监听topic，传递过来Json数据，先进行转换，再判断消息类型进行一个消费
        try {
            HouseIndexMessage message = objectMapper.readValue(content, HouseIndexMessage.class);
            switch (message.getOperation()){
                case HouseIndexMessage.INDEX:
                    this.createOrUpdateIndex(message);
                    break;
                case HouseIndexMessage.REMOVE:
                    this.removeIndex(message);
                    break;
                default:
                    logger.warn("Not support message content"+content);
                    break;
            }
        } catch (IOException e) {
            logger.error("Cannot parse json for"+content,e);
        }
    }

    private void removeIndex(HouseIndexMessage message) {
        Long houseId = message.getHouseId();
        DeleteByQueryRequestBuilder builder = DeleteByQueryAction.INSTANCE
                .newRequestBuilder(esClient)
                .filter(QueryBuilders.termQuery(HouseIndexKey.HOUSE_ID, houseId))
                .source(INDEX_NAME);

        logger.debug("Delete by query for house:"+builder);
        BulkByScrollResponse response = builder.get();
        long deleted = response.getDeleted();
        logger.debug("Delete total"+deleted);

        ServiceResult serviceResult = addressService.removeLbs(houseId);
        if (!serviceResult.isSuccess()||deleted <= 0){
            logger.warn("Did not remove data from es for response:"+response);
            //消费失败，重新加入消费队列
            this.remove(houseId,message.getRetry()+1);
        }
    }

    private void createOrUpdateIndex(HouseIndexMessage message){
        Long houseId = message.getHouseId();
        House house = houseRepository.findOne(houseId);
        if (house==null){
            logger.error("Index house {} dose not exist!",houseId);
            //数据有问题，重新查询，计数器加一
            this.index(houseId,message.getRetry()+1);
            return ;
        }
        HouseIndexTemplate indexTemplate = new HouseIndexTemplate();

        modelMapper.map(house,indexTemplate);

        HouseDetail detail = houseDetailRepository.findByHouseId(houseId);
        if (detail == null){
            // TODO 异常情况

        }

        modelMapper.map(detail,indexTemplate);

        SupportAddress city = supportAddressRepository.findByEnNameAndLevel(house.getCityEnName(), SupportAddress.Level.CITY.getValue());

        SupportAddress region = supportAddressRepository.findByEnNameAndLevel(house.getRegionEnName(), SupportAddress.Level.REGION.getValue());

        String address = city.getCnName()+region.getCnName()+house.getStreet()+house.getDistrict()+detail.getDetailAddress();

        //通过城市加地址获取经纬度
        ServiceResult<BaiduMapLocation> baiduMapLoaction = addressService.getBaiduMapLoaction(city.getCnName(), address);
        if (!baiduMapLoaction.isSuccess()){
            this.index(houseId, message.getRetry()+1);
            return;
        }
        //设置索引
        indexTemplate.setLocation(baiduMapLoaction.getResult());
        List<HouseTag> tags = tagRepository.findAllByHouseId(houseId);
        if (tags != null && !tags.isEmpty()){
            List<String> tagStrings = new ArrayList<>();
            tags.forEach(houseTag -> tagStrings.add(houseTag.getName()));
            indexTemplate.setTags(tagStrings);
        }

        SearchRequestBuilder requestBuilder = this.esClient.prepareSearch(INDEX_NAME).setTypes(INDEX_TYPE)
                .setQuery(QueryBuilders.termQuery(HouseIndexKey.HOUSE_ID, houseId));
        logger.debug(requestBuilder.toString());
        SearchResponse searchResponse = requestBuilder.get();

        long totalHit = searchResponse.getHits().getTotalHits();
        boolean success;
        if (totalHit == 0){
            //creat
            success=create(indexTemplate);
        }else  if(totalHit ==1){
            //update
            String esId = searchResponse.getHits().getAt(0).getId();
            success= update(esId,indexTemplate);
        }else {
            //不止一个，有问题，处理
            //delete & create
            success=deleteAndCreate(totalHit,indexTemplate);
        }

        //lbs数据创建  百度关闭麻点云检索功能，
//        ServiceResult serviceResult = addressService.lbsUpload(baiduMapLoaction.getResult(), house.getStreet() + house.getDistrict(), city.getCnName() + region.getCnName() + house.getStreet() + house.getDistrict(),
//                message.getHouseId(), house.getPrice(), house.getArea());
        if (!success){//||!serviceResult.isSuccess()
            //失败重新放回队列
            this.index(message.getHouseId(),message.getRetry()+1);
        }else {
            logger.debug("Index success with house"+houseId);
        }

    }
    /**
     * 将目标数据从mysql中查出来，再去es中构建
     * @param houseId
     */
    @Override
    public void index(Long houseId) {
        //走kafka消息队列
        this.index(houseId,0);
    }

    public void index(Long houseId,int retry){
        if (retry>HouseIndexMessage.MAX_RETRY){
            logger.error("Retry index times over 3 for house"+houseId+"Please check it!");
            return;
        }
        HouseIndexMessage message = new HouseIndexMessage(houseId,HouseIndexMessage.INDEX,retry);
        try {
            //kafka消息队列 通过自定义的topic将数据发送给消费者
            kafkaTemplate.send(INDEX_TOPIC,objectMapper.writeValueAsString(message));
        } catch (JsonProcessingException e) {
            logger.error("Json encode error for "+message);
        }
    }
    private boolean create(HouseIndexTemplate indexTemplate){
        if (!updateSuggest(indexTemplate)) {
            return false;
        }
        try {
            IndexResponse response = this.esClient.prepareIndex(INDEX_NAME, INDEX_TYPE)
                    .setSource(objectMapper.writeValueAsBytes(indexTemplate), XContentType.JSON).get();

            logger.debug("Create index with house:"+indexTemplate.getHouseId());
            if (response.status() == RestStatus.CREATED){
                return true;
            }else {
                return false;
            }

        } catch (JsonProcessingException e) {
            logger.error("Error to index house"+e);
            return false;
        }

    }

    private boolean  update(String esId,HouseIndexTemplate indexTemplate){
        if (!updateSuggest(indexTemplate)) {
            return false;
        }
        try {
            UpdateResponse response = this.esClient.prepareUpdate(INDEX_NAME, INDEX_TYPE,esId)
                    .setDoc(objectMapper.writeValueAsBytes(indexTemplate), XContentType.JSON).get();

            logger.debug("Update index with house:"+indexTemplate.getHouseId());
            if (response.status() == RestStatus.OK){
                return true;
            }else {
                return false;
            }

        } catch (JsonProcessingException e) {
            logger.error("Error to index house"+e);
            return false;
        }
    }
    private boolean  deleteAndCreate(long totalHit,HouseIndexTemplate indexTemplate){
        DeleteByQueryRequestBuilder builder = DeleteByQueryAction.INSTANCE
                .newRequestBuilder(esClient)
                .filter(QueryBuilders.termQuery(HouseIndexKey.HOUSE_ID, indexTemplate.getHouseId()))
                .source(INDEX_NAME);

        logger.debug("Delete by query for house:"+builder);
        BulkByScrollResponse response = builder.get();
        long deleted = response.getDeleted();
        if (deleted != totalHit){
            logger.warn("Need delete {}, but {} was deleted!",totalHit,deleted);
            return false;
        }else {
            return create(indexTemplate);
        }

    }
    @Override
    public void remove(Long houseId) {
        this.remove(houseId,0);

    }

    @Override
    public ServiceMultiResult<Long> query(RentSearch rentSerach) {
        //通过es查询对应id
        BoolQueryBuilder boolQuery = QueryBuilders.boolQuery();
        boolQuery.filter(QueryBuilders.termQuery(HouseIndexKey.CITY_EN_NAME,rentSerach.getCityEnName()));
        if (rentSerach.getRegionEnName() != null&&!"*".equals(rentSerach.getRegionEnName())){
            boolQuery.filter(
                    QueryBuilders.termQuery(HouseIndexKey.REGION_EN_NAME,rentSerach.getRegionEnName())
            );
        }
        //面积条件筛选
        RentValueBlock area = RentValueBlock.matchArea(rentSerach.getAreaBlock());
        if (!RentValueBlock.All.equals(area)){
            //范围查询
            RangeQueryBuilder rangeQueryBuilder = QueryBuilders.rangeQuery(HouseIndexKey.AREA);
            if (area.getMax() >0){
                //小于最大值  lte小于等于
                rangeQueryBuilder.lte(area.getMax());
            }
            if (area.getMin()>0){
                //大于最小值  gte大于等于
                rangeQueryBuilder.gte(area.getMin());
            }
            //组合条件
            boolQuery.filter(rangeQueryBuilder);
        }

        //价格条件筛选
        RentValueBlock price = RentValueBlock.matchPrice(rentSerach.getPriceBlock());
        if (!RentValueBlock.All.equals(price)){
            //范围查询
            RangeQueryBuilder rangeQuery = QueryBuilders.rangeQuery(HouseIndexKey.PRICE);
            if (price.getMax() >0){
                //小于最大值
                rangeQuery.lte(price.getMax());
            }
            if (price.getMin()>0){
                //大于最小值
                rangeQuery.gte(price.getMin());
            }
            //组合条件
            boolQuery.filter(rangeQuery);
        }
        //朝向条件筛选
        if (rentSerach.getDirection() >0){
            boolQuery.filter(
                    QueryBuilders.termQuery(HouseIndexKey.DIRECTION,rentSerach.getDirection())

            );
        }
        if(rentSerach.getRoom()>0){
            boolQuery.filter(
                    QueryBuilders.termQuery(HouseIndexKey.ROOM,rentSerach.getRoom())

            );
        }
        //租赁方式条件筛选
        if (rentSerach.getRentWay() > -1){
            boolQuery.filter(
                    QueryBuilders.termQuery(HouseIndexKey.RENT_WAY,rentSerach.getRentWay())

            );
        }
        HighlightBuilder highlightBuilder = new HighlightBuilder();
        if (rentSerach.getKeywords()!=null&&!rentSerach.getKeywords().isEmpty()){
            //结合业务
            //筛选  匹配关键词  搜索引擎实现  优化
            //must 必须满足的条件
//            boolQuery.should(QueryBuilders.matchQuery(HouseIndexKey.TITLE,rentSerach.getKeywords())
//                    .boost(2.0f)
//            );
            boolQuery.must(QueryBuilders.multiMatchQuery(rentSerach.getKeywords(),
                    HouseIndexKey.TITLE,
                    HouseIndexKey.TRAFFIC,
                    HouseIndexKey.DISTRICT,
                    HouseIndexKey.ROUND_SERVICE,
                    HouseIndexKey.SUBWAY_LINE_NAME,
                    HouseIndexKey.SUBWAY_STATION_NAME
            ));
        }
        SearchRequestBuilder requestBuilder = this.esClient.prepareSearch(INDEX_NAME)
                .setTypes(INDEX_TYPE)
                .setQuery(boolQuery)//查询条件
                .addSort(
                        HouseSort.getSortKey(rentSerach.getOrderBy()),//排序的关键词
                        SortOrder.fromString(rentSerach.getOrderDirection())//排序的顺序
                )
                .setFrom(rentSerach.getStart())//数据起始点
                .setSize(rentSerach.getSize())//数据大小
                 .setFetchSource(HouseIndexKey.HOUSE_ID,null);//解决返回集过大的问题，结果只拿houseid
//                .highlighter(highlightBuilder);
        logger.debug(requestBuilder.toString());
        List<Long> houseId = new ArrayList<>();
        SearchResponse response = requestBuilder.get();
        if (response.status() != RestStatus.OK){
            //结果不正常
            logger.warn("Search status is no ok for"+requestBuilder);
            return new ServiceMultiResult<>(0,houseId);
        }
        for (SearchHit hit : response.getHits()) {

            houseId.add(Longs.tryParse(String.valueOf(hit.getSource().get(HouseIndexKey.HOUSE_ID))));
        }
        return new ServiceMultiResult<>(response.getHits().totalHits,houseId);
    }

    /**
     * 补全搜索实现
     * @param prefix
     * @return
     */
    @Override
    public ServiceResult<List<String>> suggest(String prefix) {
        //设置suggest规则大小
        CompletionSuggestionBuilder suggestion = SuggestBuilders.completionSuggestion("suggest").prefix(prefix).size(5);
        SuggestBuilder suggestBuilder = new SuggestBuilder();
        suggestBuilder.addSuggestion("autocomplete",suggestion);
        //搜索
        SearchRequestBuilder requestBuilder = this.esClient.prepareSearch(INDEX_NAME)
                .setTypes(INDEX_TYPE)
                .suggest(suggestBuilder);
        logger.debug(requestBuilder.toString());

        //获取结果
        SearchResponse response = requestBuilder.get();
        Suggest suggest = response.getSuggest();
        Suggest.Suggestion result = suggest.getSuggestion("autocomplete");
        int maxSuggest = 0;
        //结果过滤
        Set<String> suggestSet = new HashSet<>();

        for (Object entry : result.getEntries()) {
            if (entry instanceof Suggest.Suggestion.Entry){
                CompletionSuggestion.Entry item = (CompletionSuggestion.Entry) entry;
                //获取的选项为空
                if (item.getOptions().isEmpty()){
                    continue;
                }
                //提供的备选项
                for (CompletionSuggestion.Entry.Option option : item.getOptions()) {
                    //排除
                    String tip = option.getText().string();
                    if (suggestSet.contains(tip)){
                        continue;
                    }
                    suggestSet.add(tip);
                    maxSuggest++;
                }
            }
            if (maxSuggest>5){
                break;
            }
        }
        ArrayList<String> strings = Lists.newArrayList(suggestSet.toArray(new String[]{}));
        return ServiceResult.of(strings);
    }

    /**
     * 聚合小区数据
     * @param cityEnName
     * @param regionEnName
     * @param district
     * @return
     */
    @Override
    public ServiceResult<Long> aggregateDistrictHouse(String cityEnName, String regionEnName, String district) {
        //聚合查询
        BoolQueryBuilder boolQuery = QueryBuilders.boolQuery()
                .filter(QueryBuilders.termQuery(HouseIndexKey.CITY_EN_NAME,cityEnName))//倒排索引精确查找 term精确查找
                .filter(QueryBuilders.termQuery(HouseIndexKey.REGION_EN_NAME,regionEnName))
                .filter(QueryBuilders.termQuery(HouseIndexKey.DISTRICT,district));

        SearchRequestBuilder requestBuilder =this.esClient.prepareSearch(INDEX_NAME)
                .setTypes(INDEX_TYPE)
                .setQuery(boolQuery)
                .addAggregation(
                        //聚合
                        AggregationBuilders.terms(HouseIndexKey.AGG_DISTRICT).field(HouseIndexKey.DISTRICT)
                ).setSize(0);
        logger.debug(requestBuilder.toString());
        SearchResponse response = requestBuilder.get();
        if (response.status() == RestStatus.OK){
           Terms terms = response.getAggregations().get(HouseIndexKey.AGG_DISTRICT);
            if (terms.getBuckets() !=null && !terms.getBuckets().isEmpty()){
                return ServiceResult.of(terms.getBucketByKey(district).getDocCount());
            }
        }else {
            logger.warn("Faild to Aggregate for"+HouseIndexKey.AGG_DISTRICT);

        }
        return ServiceResult.of(0L);
    }

    /**
     * 聚合城市区域数据
     * @param cityEnName
     * @return
     */
    @Override
    public ServiceMultiResult<HouseBuketDTO> mapAggregate(String cityEnName) {
        BoolQueryBuilder boolQuery = QueryBuilders.boolQuery();
        boolQuery.filter(QueryBuilders.termQuery(HouseIndexKey.CITY_EN_NAME,cityEnName));

        AggregationBuilder aggBulider = AggregationBuilders.terms(HouseIndexKey.AGG_REGION)
                .field(HouseIndexKey.REGION_EN_NAME);
        SearchRequestBuilder requestBuilder = this.esClient.prepareSearch(INDEX_NAME)
                .setTypes(INDEX_TYPE)
                .setQuery(boolQuery)
                .addAggregation(aggBulider);

        logger.debug(requestBuilder.toString());

        SearchResponse response = requestBuilder.get();

        List<HouseBuketDTO> buketDTOS = new ArrayList<>();
        if (response.status() != RestStatus.OK){
            logger.warn("Aggregate status is not ok for" + requestBuilder);
            return new ServiceMultiResult<>(0, buketDTOS);
        }
        Terms terms = response.getAggregations().get(HouseIndexKey.AGG_REGION);
        for (Terms.Bucket bucket : terms.getBuckets()) {
            buketDTOS.add(new HouseBuketDTO(bucket.getKeyAsString(),bucket.getDocCount()));
        }
        return new ServiceMultiResult<>(response.getHits().getTotalHits(),buketDTOS);
    }



    @Override
    public ServiceMultiResult<Long> mapQuery(String cityEnName, String orderBy, String orederDirection, int start, int size) {
            BoolQueryBuilder boolQuery = QueryBuilders.boolQuery();
            boolQuery.filter(QueryBuilders.termQuery(HouseIndexKey.CITY_EN_NAME,cityEnName));
            SearchRequestBuilder requestBuilder = this.esClient.prepareSearch(INDEX_NAME)
                .setTypes(INDEX_TYPE)
                .setQuery(boolQuery)
                .addSort(HouseSort.getSortKey(orderBy), SortOrder.fromString(orederDirection))
                .setFrom(start)
                .setSize(size);


        List<Long> houseIds = new ArrayList<>();
        SearchResponse response = requestBuilder.get();
        if (response.status()!= RestStatus.OK){
            logger.warn("Search status is not ok for"+requestBuilder);
            return new ServiceMultiResult<>(0,houseIds);

        }
        for (SearchHit hit : response.getHits()) {
            houseIds.add(Longs.tryParse(String.valueOf(hit.getSource().get(HouseIndexKey.HOUSE_ID))));
        }
        return new ServiceMultiResult<>(response.getHits().getTotalHits(),houseIds);
    }

    //地图动作触发时进行查询，通过传过来的四个点位置查询包含在其中的
    @Override
    public ServiceMultiResult<Long> mapQuery(MapSearch mapSearch) {
        BoolQueryBuilder boolQuery = QueryBuilders.boolQuery();
        boolQuery.filter(QueryBuilders.termQuery(HouseIndexKey.CITY_EN_NAME,mapSearch.getCityEnName()));
        boolQuery.filter(QueryBuilders.geoBoundingBoxQuery("location").setCorners(
                new GeoPoint(mapSearch.getLeftLatitude(),mapSearch.getLeftLongitude()),//先左后右，精确位置查询
                new GeoPoint(mapSearch.getRightLatitude(),mapSearch.getRightLongitude())

        ));
        SearchRequestBuilder requestBuilder = this.esClient.prepareSearch(INDEX_NAME)
                .setTypes(INDEX_TYPE)
                .setQuery(boolQuery)
                .addSort(HouseSort.getSortKey(mapSearch.getOrderBy()),
                        SortOrder.fromString(mapSearch.getOrderDirection()))
                .setFrom(mapSearch.getStart())
                .setSize(mapSearch.getSize());
        List<Long> houseIds = new ArrayList<>();
        SearchResponse response = requestBuilder.get();
        if (response.status()!= RestStatus.OK){
            logger.warn("Search status is not ok for"+requestBuilder);
            return new ServiceMultiResult<>(0,houseIds);

        }
        for (SearchHit hit : response.getHits()) {
            houseIds.add(Longs.tryParse(String.valueOf(hit.getSource().get(HouseIndexKey.HOUSE_ID))));
        }
        return new ServiceMultiResult<>(response.getHits().getTotalHits(),houseIds);
    }

    @Override
    public ServiceMultiResult<Long> mapQuery(MapSearchDistrict mapSearch) {
        BoolQueryBuilder boolQuery = QueryBuilders.boolQuery();
        boolQuery.filter(QueryBuilders.termQuery(HouseIndexKey.CITY_EN_NAME,mapSearch.getCityEnName()));
        boolQuery.filter(QueryBuilders.termQuery(HouseIndexKey.DISTRICT,mapSearch.getDistrict()));


        SearchRequestBuilder requestBuilder = this.esClient.prepareSearch(INDEX_NAME)
                .setTypes(INDEX_TYPE)
                .setQuery(boolQuery)
                .addSort(HouseSort.getSortKey(mapSearch.getOrderBy()),
                        SortOrder.fromString(mapSearch.getOrderDirection()))
                .setFrom(mapSearch.getStart())
                .setSize(mapSearch.getSize());
        List<Long> houseIds = new ArrayList<>();
        SearchResponse response = requestBuilder.get();
        if (response.status()!= RestStatus.OK){
            logger.warn("Search status is not ok for"+requestBuilder);
            return new ServiceMultiResult<>(0,houseIds);

        }
        for (SearchHit hit : response.getHits()) {
            houseIds.add(Longs.tryParse(String.valueOf(hit.getSource().get(HouseIndexKey.HOUSE_ID))));
        }
        return new ServiceMultiResult<>(response.getHits().getTotalHits(),houseIds);
    }
    @Override
    public ServiceMultiResult<HouseBuketDTO> mapAggregateDistrict(SearchDistrict searchDistrict) {
        BoolQueryBuilder boolQuery = QueryBuilders.boolQuery();
        boolQuery.filter(QueryBuilders.termQuery(HouseIndexKey.CITY_EN_NAME,searchDistrict.getCityEnName()));
//        .filter(QueryBuilders.geoBoundingBoxQuery("location").setCorners(
//                new GeoPoint(searchDistrict.getLeftLatitude(),searchDistrict.getLeftLongitude()),//先左后右，精确位置查询
//                new GeoPoint(searchDistrict.getRightLatitude(),searchDistrict.getRightLongitude())
//        ));

        AggregationBuilder aggBulider = AggregationBuilders.terms(HouseIndexKey.AGG_DISTRICT)
                .field(HouseIndexKey.DISTRICT);
        SearchRequestBuilder requestBuilder = this.esClient.prepareSearch(INDEX_NAME)
                .setTypes(INDEX_TYPE)
                .setQuery(boolQuery)
                .addAggregation(aggBulider);

        logger.debug(requestBuilder.toString());

        SearchResponse response = requestBuilder.get();

        List<HouseBuketDTO> buketDTOS = new ArrayList<>();
        if (response.status() != RestStatus.OK){
            logger.warn("Aggregate status is not ok for" + requestBuilder);
            return new ServiceMultiResult<>(0, buketDTOS);
        }
        Terms terms = response.getAggregations().get(HouseIndexKey.AGG_DISTRICT);
        for (Terms.Bucket bucket : terms.getBuckets()) {
            buketDTOS.add(new HouseBuketDTO(bucket.getKeyAsString(),bucket.getDocCount()));
        }
        return new ServiceMultiResult<>(response.getHits().getTotalHits(),buketDTOS);

    }

    private boolean updateSuggest(HouseIndexTemplate indexTemplate){
        //创建索引时使用
        //更新补全搜索字段
        //对分词接口进行请求
        AnalyzeRequestBuilder requestBuilder = new AnalyzeRequestBuilder(this.esClient, AnalyzeAction.INSTANCE,INDEX_NAME,indexTemplate.getTitle(),indexTemplate.getLayoutDesc(),indexTemplate.getRoundService(),
                indexTemplate.getDescription(),indexTemplate.getSubwayLineName(),
                indexTemplate.getSubwayStationName());
        requestBuilder.setAnalyzer("ik_smart");
        //获取使用ik_smart规则分好的词
        AnalyzeResponse response = requestBuilder.get();
        List<AnalyzeResponse.AnalyzeToken> tokens = response.getTokens();
        if (tokens == null){
            logger.warn("Can not analyze token for house"+indexTemplate.getHouseId());
            return false;
        }
        List<HouseSuggest> suggests = new ArrayList<>();
        for (AnalyzeResponse.AnalyzeToken token : tokens) {
            //排除数字类型或者小于两个字符的
            if ("<NUM>".equals(token.getType())&&token.getTerm().length()<2){
                continue;
            }
            //封装 默认权重相同都是10
            HouseSuggest suggest = new HouseSuggest();
            suggest.setInput(token.getTerm());
            suggests.add(suggest);

        }
        //定制化数据
        HouseSuggest suggest = new HouseSuggest();
        suggest.setInput(indexTemplate.getDistrict());
        suggests.add(suggest);
        indexTemplate.setSuggest(suggests);
        return true;

    }


    private void remove(Long houseId,int retry){
        if (retry  > HouseIndexMessage.MAX_RETRY){
            logger.error("Retry remove times over 3 for house"+houseId+"Please check it");
            return;
        }
        HouseIndexMessage message = new HouseIndexMessage(houseId,HouseIndexMessage.REMOVE,retry);
        try {
            this.kafkaTemplate.send(INDEX_TOPIC,objectMapper.writeValueAsString(message));
        } catch (JsonProcessingException e) {
            logger.error("Cannot encode json for"+message,e);
        }
    }
}
